package org.baojie.raft.box;

import lombok.Data;
import org.baojie.raft.connector.Connector;
import org.baojie.raft.connector.RaftChannel;
import org.baojie.raft.delay.MessageCache;
import org.baojie.raft.group.RaftGroup;
import org.baojie.raft.message.*;
import org.baojie.raft.node.CoreInNode;
import org.baojie.raft.node.NodeName;
import org.baojie.raft.node.type.Follower;
import org.baojie.raft.RaftNode;
import org.baojie.raft.node.type.Candidate;
import org.baojie.raft.node.type.Leader;
import org.baojie.raft.node.type.Role;
import org.baojie.raft.store.NodeStore;
import org.baojie.raft.util.current.LocalSleep;
import org.baojie.raft.util.current.RaftFactory;
import org.baojie.raft.util.current.RaftThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static org.baojie.raft.node.type.Role.Name.*;

@Data
public abstract class AbstractBox implements Box {

    private static final Logger log = LoggerFactory.getLogger(AbstractBox.class);

    private final ThreadLocalRandom random = ThreadLocalRandom.current();

    private final MessageCache<String, RaftMsgSignal> waitVoteSignals = MessageCache.create(4096);

    private final RaftThreadPool waitVotePool = new RaftThreadPool(
            16,
            256,
            60,
            TimeUnit.SECONDS,
            new SynchronousQueue<>(),
            RaftFactory.create("deal vote response in box", false));

    private final AtomicBoolean stop = new AtomicBoolean(true);

    private final List<Future<?>> futures = new ArrayList<>(64);

    private final RaftGroup group;
    private final RaftNode self;
    private final Connector connector;
    private final NodeStore store;
    private final NodeName name;

    public AbstractBox(NodeName name, NodeStore store, RaftGroup group) {
        this.name = name;
        this.store = store;
        this.group = group;
        this.connector = new RaftChannel(this);
        this.self = new RaftNode(this);
    }

    @Override
    public NodeName name() {
        return name;
    }

    @Override
    public NodeStore store() {
        return store;
    }

    @Override
    public RaftGroup group() {
        return group;
    }

    @Override
    public void start() {
        if (stop.compareAndSet(true, false)) {
            if (!connector.start()) {
                throw new Error("connector start error ...");
            }
            this.self.start();
        }
    }

    @Override
    public void stop() {
        if (stop.compareAndSet(false, true)) {
            connector.stop();
            try {
                store.close();
            } catch (IOException e) {

            }
            self.stop();
            for (Future<?> fu : futures) {
                fu.cancel(true);
            }
        }
    }

    @Override
    public void putPing(RaftMessage ping) {
        if (null == ping) {
            return;
        } else {
            boolean suc = self.putPing(ping);
            if (!suc) {
                //dealHeartBeatPing(ping);
            }
        }
    }

    @Override
    public RaftMsgSignal touchStartVote(String msgId) {
        return waitVoteSignals.get(msgId);
    }

    @Override
    public void doNodeVote(Role oldRole, int oldStamp, String msgId) {
        int test = waitVoteSignals.size();
        if (test > 0) {
            if (test > 1) {
                log.error("wait vote signals can not be bigger than 1, cache size = " + test);
            }
            return;
        }
        int needToSend = group.exceptSelf().size();
        RaftMsgSignal signal = new RaftMsgSignal(msgId, needToSend, oldStamp);
        long min = CoreInNode.Meta.min_elc_time.millis();
        if (null != waitVoteSignals.putIfAbsent(msgId, signal, min, TimeUnit.MILLISECONDS)) {
            log.error("缓存选举信号标志失败");
            return;
        }
        for (; ; ) {
            Role role = self.role();
            int stamp = self.roleStamp();
            if (stamp != oldStamp || role.type() == LEADER) {
                if (signal.setCancel()) {
                    signal.releaseAll();
                }
                waitVoteSignals.remove(msgId);
                return;
            }
            if (self.touch()) {
                try {
                    role = self.role();
                    stamp = self.roleStamp();
                    if (stamp != oldStamp || role.type() == LEADER) {
                        if (signal.setCancel()) {
                            signal.releaseAll();
                        }
                        waitVoteSignals.remove(msgId);
                        return;
                    }
                    // 不为 leader 就剩下两个状态了
                    final Candidate candidate;
                    if (role.type() == CANDIDATE) {
                        // 继承原来的投票数
                        candidate = new Candidate(role.store(), role.votedCount(), role.getTerm() + 1);
                    } else {
                        // 如果是 follower 投票数为 0
                        candidate = new Candidate(role.store(), 0, role.getTerm() + 1);
                    }
                    if (self.stampSetRole(oldRole, candidate, oldStamp)) {
                        // 如果放入成功那么stamp已经自动 +1 了
                        startVote4Candidate(candidate, msgId, signal, needToSend);
                    }
                } finally {
                    self.unTouch();
                }
                return;
            } else {
                if (stop.get()) {
                    return;
                } else {
                    LocalSleep.safePoint();
                }
            }
        }
    }

    private void startVote4Candidate(Candidate candidate, String msgId, RaftMsgSignal signal, int needToSend) {
        VoteRequest vote = buildVote(candidate);
        // 设置需要等待多少信号返回
        AtomicInteger count = new AtomicInteger(needToSend);
        final int i;
        // 计算等待信号的 worker, 最大值不能超过64, 可以调整 增大或者减小
        if (count.get() >= 64) {
            i = 64;
        } else {
            i = count.get();
        }
        // 使用通信组件发送消息
        List<NodeName> nodes = group.exceptSelf();
        for (NodeName remote : nodes) {
            RaftMessage request = new RaftMessage(msgId, vote, self.name(), remote);
            // 放入待发消息队列
            sendMessage(request);
        }
        // 创建worker去等待状态
        for (int j = 0; j < i; j++) {
            WaitForVoteResponse wait = new WaitForVoteResponse(count, signal);
            waitVotePool.submit(wait);
        }
    }

    private VoteRequest buildVote(Candidate candidate) {
        final VoteRequest vote = new VoteRequest();
        vote.setTerm(candidate.getTerm());
        vote.setCandidateName(name);
        vote.setLastLogIndex(0);
        vote.setLastLogTerm(0);
        return vote;
    }

    // 处理 收到心跳
    @Override
    public void dealHeartBeatPing(RaftMessage msg, Role oldRole, int oldStamp) {
        HeartBeatPing ping = (HeartBeatPing) msg.getMessage();
        String msgId = msg.getMsgId();
        for (; ; ) {
            if (self.touch()) {
                try {
                    int curStamp = self.roleStamp();
                    Role role = self.role();
                    if (curStamp != oldStamp) {
                        log.warn("收到心跳后发现版本已经变化，不作处理，直接退出");
                        return;
                    }
                    if (ping.getTerm() < role.getTerm()) {
                        HeartBeatPong pong = new HeartBeatPong();
                        pong.setSuccess(false);
                        pong.setTerm(role.getTerm());
                        RaftMessage request = new RaftMessage(msgId, pong, self.name(), msg.from());
                        // 放入待发消息队列
                        sendMessage(request);
                        // 锁住之后再次检查任期仍然符合要求处理完成之后直接返回
                        return;
                    }

                    NodeStore store = role.store();
                    if (ping.getTerm() > role.getTerm()) {

                        // 设置node状态
                        Follower follower = new Follower(null, ping.getLeader(), store, ping.getTerm());
                        boolean suc = self.stampSetRole(role, follower, curStamp);
                        if (!suc) {
                            log.error("收到 ping 后发现set出错，出现严重异常");
                        }

                        cancelSignalWithOutMsgId();

                        HeartBeatPong pong = new HeartBeatPong();
                        pong.setSuccess(true);
                        pong.setTerm(ping.getTerm());
                        RaftMessage request = new RaftMessage(msgId, pong, self.name(), msg.from());
                        // 放入待发消息队列
                        sendMessage(request);
                        // 锁住之后再次检查任期
                        // 仍然符合要求处理完成之后直接返回
                        return;
                    }

                    Follower follower = null;
                    RaftMsgSignal signal = null;
                    HeartBeatPong pong = null;
                    RaftMessage request = null;
                    boolean suc = false;

                    cancelSignalWithOutMsgId();

                    switch (role.type()) {
                        case FOLLOWER:
                            // 设置node状态
                            follower = new Follower(role.votedFor(), ping.getLeader(), store, ping.getTerm());
                            suc = self.stampSetRole(role, follower, curStamp);
                            if (!suc) {
                                log.error("收到 ping 后， 根据角色follower, set follower出错，出现严重异常");
                            }

                            pong = new HeartBeatPong();
                            pong.setSuccess(true);
                            pong.setTerm(ping.getTerm());
                            request = new RaftMessage(msgId, pong, self.name(), msg.from());
                            sendMessage(request);
                            return;

                        case CANDIDATE:
                            // 设置node状态
                            follower = new Follower(null, ping.getLeader(), store, ping.getTerm());
                            suc = self.stampSetRole(role, follower, curStamp);
                            if (!suc) {
                                log.error("收到 ping 后， 根据角色 candidate, set follower 出错，出现严重异常");
                            }

                            pong = new HeartBeatPong();
                            pong.setSuccess(true);
                            pong.setTerm(ping.getTerm());
                            request = new RaftMessage(msgId, pong, self.name(), msg.from());
                            sendMessage(request);

                            return;

                        case LEADER:

                            pong = new HeartBeatPong();
                            pong.setSuccess(false);
                            pong.setTerm(role.getTerm());
                            request = new RaftMessage(msgId, pong, self.name(), msg.from());
                            // 放入待发消息队列
                            sendMessage(request);
                            return;

                        default:
                            log.error("接受 ping 后，居然检测出位置role = " + role.type());
                            return;
                    }
                } finally {
                    self.unTouch();
                }
            } else {
                if (stop.get()) {
                    return;
                } else {
                    LocalSleep.safePoint();
                }
            }
        }
    }


    // 处理  请求投票的响应
    private final class WaitForVoteResponse implements Runnable {

        private final AtomicInteger count;
        private final RaftMsgSignal signal;

        public WaitForVoteResponse(AtomicInteger count, RaftMsgSignal signal) {
            this.count = count;
            this.signal = signal;
        }

        @Override
        public void run() {
            for (; ; ) {
                if (stop.get()) {
                    return;
                } else {
                    int index = count.decrementAndGet();
                    if (index >= 0) {
                        boolean suc = tryAcquire();
                        if (signal.isCanceled()) {
                            // 如果取消 那么直接删除等待信号的缓存
                            doClean();
                            return;
                        }
                        if (suc) {
                            if (signal.isCanceled()) {
                                doClean();
                                return;
                            } else {
                                dealVoteResponse(index);
                            }
                        }
                        if (signal.isCanceled()) {
                            doClean();
                            return;
                        }
                    } else {
                        doClean();
                        return;
                    }
                }
            }
        }

        private void doClean() {
            try {
                if (signal.setCancel()) {
                    signal.releaseAll();
                }
            } finally {
                waitVoteSignals.remove(signal.getMsgId());
            }
        }

        // 处理 requestVote响应
        private void dealVoteResponse(int index) {
            VoteResponse tmp = signal.getEntity(index);
            if (null == tmp) {
                if (0 == index) {
                    doClean();
                }
                return;
            }
            Role role = self.role();
            int curStamp = self.roleStamp();
            // signal中的stamp是set成功之前的
            // 一旦set成功那么现在的stamp是加一之后的
            int candidateStamp = signal.getOldStamp() + 1;
            // 说明现在的版本已经不是需要处理的版本了
            if (curStamp != candidateStamp || role.type() == LEADER) {
                doClean();
                return;
            }
            for (; ; ) {
                if (self.touch()) {
                    try {
                        role = self.role();
                        curStamp = self.roleStamp();
                        if (curStamp != candidateStamp || role.type() == LEADER) {
                            doClean();
                            return;
                        }
                        NodeStore store = role.store();
                        if (tmp.getTerm() > role.getTerm()) {

                            Follower follower = new Follower(null, null, store, tmp.getTerm());
                            boolean suc = self.stampSetRole(role, follower, curStamp);
                            if (!suc) {
                                log.error("一件不可能的事情，在加锁的情况下不会设置不成功，设置follower");
                            }
                            doClean();
                            return;
                        }

                        if (role.type() != CANDIDATE) {
                            if (index == 0) {
                                doClean();
                            }
                            return;
                        }

                        if (tmp.getTerm() < role.getTerm() || !tmp.isVoteGranted()) {
                            if (0 == index) {
                                doClean();
                            }
                            return;
                        }

                        Candidate candidate = (Candidate) role;
                        candidate.addOneCount();

                        int votes = candidate.votedCount();
                        int half = group.half();
                        if (votes > half) {
                            Leader leader = new Leader(store, role.getTerm());
                            boolean suc = self.stampSetRole(role, leader, curStamp);
                            if (!suc) {
                                log.error("一件不可能的事情，在加锁的情况下不会设置不成功，设置 leader");
                            }
                            // 取消信号
                            doClean();
                            // 立即发送心跳信号
                            self.heartBeatOthers(leader);
                            return;
                        }
                        if (index == 0) {
                            // 重置为 1 票
                            Candidate newer = new Candidate(role.store(), 1, role.getTerm());
                            boolean suc = self.stampSetRole(role, newer, curStamp);
                            if (!suc) {
                                log.error("一件不可能的事情，在加锁的情况下不会设置不成功，设置 candidate");
                            }
                            doClean();
                            return;
                        }

                    } finally {
                        self.unTouch();
                    }
                    return;
                } else {
                    if (stop.get()) {
                        return;
                    } else {
                        if (signal.isCanceled()) {
                            doClean();
                            return;
                        } else {
                            LocalSleep.safePoint();
                        }
                    }
                }
            }
        }

        private boolean tryAcquire() {
            return signal.acquireOne(CoreInNode.Meta.max_elc_time.millis(), TimeUnit.MILLISECONDS);
        }

    }


    // 收到 请求投票 的消息
    @Override
    public void receiveVoteRequest(RaftMessage request) {
        String msgId = request.getMsgId();
        VoteRequest vote = (VoteRequest) request.getMessage();
        for (; ; ) {
            if (self.touch()) {
                try {
                    Role role = self.role();
                    int stamp = self.roleStamp();
                    // 请求者的任期小于当前节点任期
                    // 不投票
                    if (vote.getTerm() < role.getTerm()) {
                        VoteResponse response = new VoteResponse();
                        response.setTerm(role.getTerm());
                        response.setVoteGranted(false);
                        RaftMessage message = new RaftMessage(msgId, response, self.name(), request.from());
                        // 待发送消息放入队列
                        sendMessage(message);
                        log.warn("接收到一个小任期的请求这信号, from = " + request + ", local = " + role);
                        return;
                    }

                    if (vote.getTerm() > role.getTerm()) {
                        NodeName candidate = vote.getCandidateName();
                        Follower follower = new Follower(candidate, null, role.store(), vote.getTerm());
                        if (self.stampSetRole(role, follower, stamp)) {

                            cancelSignalWithOutMsgId();

                            VoteResponse response = new VoteResponse();
                            response.setTerm(vote.getTerm());
                            response.setVoteGranted(true);
                            RaftMessage message = new RaftMessage(msgId, response, self.name(), request.from());
                            // 待发送消息放入队列
                            sendMessage(message);
                            log.warn("接收到比自己任期大的请求投票，已经变化成 follower");
                        } else {
                            log.error("接收到请求投票，发现对方任期较大，但是设置role居然出错，不可能发生的");
                        }
                        return;
                    }

                    boolean voteForCandidate = true;

                    switch (role.type()) {
                        case CANDIDATE:
                        case LEADER:
                            VoteResponse response = new VoteResponse();
                            response.setTerm(role.getTerm());
                            response.setVoteGranted(false);
                            RaftMessage message = new RaftMessage(msgId, response, self.name(), request.from());
                            // 待发送消息放入队列
                            sendMessage(message);
                            return;
                        case FOLLOWER:
                            NodeName voteFor = role.votedFor();
                            if ((null == voteFor && voteForCandidate) || voteFor.getName().equals(vote.getCandidateName().getName())) {

                                NodeName candidate = vote.getCandidateName();
                                Follower follower = new Follower(candidate, null, role.store(), vote.getTerm());
                                boolean suc = self.stampSetRole(role, follower, stamp);
                                if (!suc) {
                                    log.error("接收到请求投票，使用版本的加锁set居然失败，严重异常");
                                }

                                cancelSignalWithOutMsgId();

                                VoteResponse resp = new VoteResponse();
                                resp.setTerm(vote.getTerm());
                                resp.setVoteGranted(true);
                                RaftMessage vo = new RaftMessage(msgId, resp, self.name(), request.from());
                                // 待发送消息放入队列
                                sendMessage(vo);
                            } else {
                                VoteResponse resp = new VoteResponse();
                                resp.setTerm(vote.getTerm());
                                resp.setVoteGranted(false);
                                RaftMessage vo = new RaftMessage(msgId, resp, self.name(), request.from());
                                // 待发送消息放入队列
                                sendMessage(vo);
                            }
                            return;
                        default:
                            log.error("居然检测出位置role = " + role.type());
                            return;
                    }
                } finally {
                    self.unTouch();
                }
            } else {
                if (stop.get()) {
                    return;
                } else {
                    LocalSleep.safePoint();
                }
            }
        }
    }

    private void cancelSignalWithOutMsgId() {
        // 取消所有定时
        List<String> keys = waitVoteSignals.touchKeys();
        int ks = keys.size();
        if (ks > 0) {
            if (ks > 1) {
                log.error("等待选举信号的缓存信号量居然大于1, size = " + ks + ", 出现严重异常");
            }
            for (String key : keys) {
                RaftMsgSignal signal = waitVoteSignals.get(key);
                if (null != signal) {
                    if (signal.setCancel()) {
                        signal.releaseAll();
                    }
                    waitVoteSignals.remove(key);
                }
            }
        }
    }

    @Override
    public void dealHeartBeatPong(RaftMessage message) {
        HeartBeatPong pong = (HeartBeatPong) message.getMessage();
        Role role = self.role();
        if (pong.getTerm() > role.getTerm()) {
            for (; ; ) {
                if (self.touch()) {
                    try {
                        role = self.role();
                        int stamp = self.roleStamp();
                        if (pong.getTerm() <= role.getTerm()) {
                            break;
                        }
                        Follower follower = new Follower(null, null, role.store(), pong.getTerm());
                        boolean suc = self.stampSetRole(role, follower, stamp);
                        if (!suc) {
                            log.error("接收到pong回应，使用版本的加锁set居然失败，严重异常");
                        }
                        cancelSignalWithOutMsgId();
                        return;
                    } finally {
                        self.unTouch();
                    }
                } else {
                    if (stop.get()) {
                        return;
                    } else {
                        LocalSleep.safePoint();
                    }
                }
            }
        }
    }

    @Override
    public void sendMessage(RaftMessage message) {
        connector.sendMessage(message);
    }

    @Override
    public String toString() {
        return "AbstractBox{" +
                "self=" + self +
                ", group=" + group +
                ", connector=" + connector +
                ", store=" + store +
                '}';
    }

}
